﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;

using MicroMq.Core.Helper;

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;

using Peer;
using Peer.Define;

namespace MicroMq.Core
{
    public class MicroMqFactory : IDisposable
    {
        private readonly PeerClient peerClient;
        private readonly ILogger<MicroMqFactory> logger;
        private readonly ConcurrentDictionary<string, List<Action<byte[]>>> consumers = new ConcurrentDictionary<string, List<Action<byte[]>>>();

        private readonly IServiceScope serviceScope;

        public MicroMqFactory(ILogger<MicroMqFactory> logger, IOptions<MicroMqOptions> options, IServiceProvider serviceProvider)
        {
            this.serviceScope = serviceProvider.CreateScope();
            peerClient = new PeerClient(options.Value.GetCipher(), options.Value.GetEndPoint());
            peerClient.OnReceive += PeerClient_OnReceive;
            peerClient.OnError += PeerClient_OnError;
            peerClient.OnConnected += PeerClient_OnConnected;
            peerClient.OnDisconnect += PeerClient_OnDisconnect;

            this.logger = logger;
        }

        public void Init()
        {
            if (!peerClient.IsConnected)
            {
                peerClient.Connect();
            }
        }

        private void PeerClient_OnConnected(object sender, PeerStream e)
        {
            logger.LogDebug("MicroMq Connected");
            var types = AssemblyHelper.GetTypes<ConsumerAttribute>(Assembly.GetEntryAssembly());
            foreach (var type in types)
            {
                InitConsumers(type);
                InitDispatchers(type);
            }
        }

        private void InitDispatchers(Type type)
        {
            var dispatchers = AssemblyHelper.GetMethods<DispatcherAttribute>(type);
            foreach (var dispatcher in dispatchers)
            {
                var taskName = GetTaskName(dispatcher.Item1, dispatcher.Item2.TaskName);
                try
                {
                    DispatcherAsync(AssemblyHelper.Proxy(serviceScope.ServiceProvider, dispatcher.Item1), taskName, dispatcher.Item2.CronExpression).GetAwaiter().GetResult();
                    logger.LogDebug($"MicroMq dispatch task: {taskName}, cron: {dispatcher.Item2.CronExpression}");
                }
                catch (Exception ex)
                {
                    logger.LogError($"MicroMq dispatch task: {taskName}, cron: {dispatcher.Item2.CronExpression}, {ex.Message}");
                }
            }
        }
        private void InitConsumers(Type type)
        {
            var consumers = AssemblyHelper.GetMethods<ListenAttribute>(type);
            foreach (var consumer in consumers)
            {
                try
                {
                    SubscribeAsync(AssemblyHelper.Proxy(serviceScope.ServiceProvider, consumer.Item1), consumer.Item2.Topic, consumer.Item2.GroupId).GetAwaiter().GetResult();
                    logger.LogDebug($"MicroMq listen topic: {consumer.Item2.Topic}, groupId: {consumer.Item2.GroupId}");
                }
                catch (Exception ex)
                {
                    logger.LogError($"MicroMq listen topic: {consumer.Item2.Topic}, groupId: {consumer.Item2.GroupId}, {ex.Message}");
                }
            }

        }

        private string GetTaskName(MethodInfo method, string taskName)
        {
            if (!string.IsNullOrEmpty(taskName)) return method.ReflectedType.Name + ":" + taskName;
            return method.ReflectedType.Name + ":" + method.Name;
        }

        private void PeerClient_OnDisconnect(object sender, PeerStream e)
        {
            logger.LogDebug("MicroMq Disconnect");
        }

        private void PeerClient_OnError(object sender, Exception e)
        {
            logger.LogError(e, "MicroMq  Error");
        }


        /// <summary>
        /// 订阅消息
        /// </summary>
        /// <param name="topicId"></param>
        /// <param name="groupId"></param>
        /// <returns></returns>
        public async Task SubscribeAsync(Action<byte[]> consumer, string topic, string groupId)
        {
            var response = await peerClient.SendWithResponseAsync(MqBuilder.Subscribe(topic, groupId));
            response.GetBytes();
            consumers.GetOrAdd(topic, k => new List<Action<byte[]>>())
                .Add(consumer);
        }

        public async Task DispatcherAsync(Action<byte[]> consumer, string taskName, string cronExpression)
        {
            var response = await peerClient.SendWithResponseAsync(MqBuilder.Dispatcher(taskName, cronExpression));
            response.GetBytes();
            consumers.GetOrAdd(taskName, k => new List<Action<byte[]>>())
                .Add(consumer);
        }

        private PeerResponse PeerClient_OnReceive(PeerStream peer, PeerRequest request)
        {
            var readBuffer = ByteReadBuffer.Wrap(StreamToBytes(request.Data));
            var topicId = readBuffer.ReadUnknownString();
            var data = readBuffer.ReadEnd();
            if (consumers.TryGetValue(topicId, out var cus))
            {
                foreach (var consumer in cus)
                {
                    TryInvoke(consumer, data);
                }
            }
            return request.CreateResponse();
        }

        private Span<byte> StreamToBytes(Stream stream)
        {
            if (stream is MemoryStream memoryStream)
            {
                return memoryStream.ToArray();
            }
            stream.Position = 0L;
            using MemoryStream memoryStream2 = new MemoryStream();
            stream.CopyTo(memoryStream2);
            return memoryStream2.ToArray();
        }

        private void TryInvoke(Action<byte[]> consumer, byte[] data)
        {
            try
            {
                consumer.Invoke(data);
            }
            catch (Exception e)
            {
                logger.LogError(e, "MicroMq  Error");
            }
        }

        public MicroProducer GetProducer()
        {
            if (!peerClient.IsConnected)
            {
                peerClient.Connect();
            }
            return new MicroProducer(peerClient);
        }

        public void Dispose()
        {
            serviceScope.Dispose();
        }
    }
}
